package com.jackstraw.jack_base.mq.rocketmq.producer.impl;

import cn.hutool.core.lang.UUID;
import com.jackstraw.jack_base.mq.common.aspect.annotation.ServerNameCut;
import com.jackstraw.jack_base.mq.common.entity.MQSendResult;
import com.jackstraw.jack_base.mq.common.entity.enums.DelayMessageLevelEnum;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMqMessage;
import com.jackstraw.jack_base.mq.rocketmq.bean.listener.ProducerTxService;
import com.jackstraw.jack_base.mq.rocketmq.entity.User;
import com.jackstraw.jack_base.mq.rocketmq.producer.BaseProducer;
import com.jackstraw.jack_base.mq.rocketmq.producer.ProducerServer;
import com.jackstraw.jack_base.mq.rocketmq.producer.ProducerServer01;
import com.jackstraw.jack_base.mq.rocketmq.producer.listener.impl.Default2ProducerTxServiceImpl;
import com.jackstraw.jack_base.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Date;
import java.util.concurrent.TimeUnit;

@Slf4j
@ServerNameCut(serverName = "cluster220")
public class ProducerServerImpl01 implements ProducerServer01 {

    @Autowired
    private BaseProducer baseProducer;

    @Override
    public void all() {
        this.send();
        this.sendAndConvert();
        this.sendMessageInTransaction();
        this.sendSyncDelayLevel();
        this.sendSync();
        this.sendAsyncOrderly();
    }

    @Override
    public void send() {
        RocketMqMessage<String> ordinaryTopicMessage = new RocketMqMessage<>("ordinaryTopic", "普通消息");
        baseProducer.send(ordinaryTopicMessage);

        RocketMqMessage<String> broadcastTopicMessage = new RocketMqMessage<>("broadcastTopic", "广播消息");
        baseProducer.send(broadcastTopicMessage);

        RocketMqMessage<String> clusterTopicMessage = new RocketMqMessage<>("clusterTopic", "集群消息");
        baseProducer.send(clusterTopicMessage);

    }

    @Override
    public MQSendResult sendAndConvert() {
        User user = new User();
        user.setUserName("序列化消息");
        user.setAge("18");

        RocketMqMessage<User> convertTopicMessage = new RocketMqMessage<>("convertTopic", user);

        MQSendResult mqSendResult = baseProducer.sendAndConvert(convertTopicMessage, User.class);

        return mqSendResult;
    }

    @Override
    public MQSendResult sendSync() {
        RocketMqMessage<String> syncTopicMessage = new RocketMqMessage<>("syncTopic", "异步回调");
        MQSendResult<SendResult> sendResult = baseProducer.sendSync(syncTopicMessage, String.class);

        RocketMqMessage<String> syncCollectionTopicMessage = new RocketMqMessage<>("syncCollectionTopic", "异步回调Collection");
        MQSendResult<SendResult> sendSyncCollection = baseProducer.sendSyncCollection(syncCollectionTopicMessage, String.class);


        return null;
    }

    @Override
    public MQSendResult sendAsyncOrderly() {

        SendCallback sendCallback = new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("顺序消息发送成功:{}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.info("顺序消息发送失败:{}", e.getMessage());
            }
        };
        RocketMqMessage<String> asyncOrderlyTopicMessage = new RocketMqMessage<>("asyncOrderlyTopic", "顺序消息");
        asyncOrderlyTopicMessage.setSendCallback(sendCallback);
        MQSendResult mqSendResult = null;

        for (int i = 0; i < 100; i++) {
            asyncOrderlyTopicMessage.setHashKey(UUID.fastUUID().toString());
            asyncOrderlyTopicMessage.setMessage("顺序消息".concat(String.valueOf(i)));
            baseProducer.sendAsyncOrderly(asyncOrderlyTopicMessage, String.class);

        }
        return mqSendResult;
    }

    @Override
    public MQSendResult<SendResult> sendSyncDelayLevel() {
        RocketMqMessage<String> syncDelayLevelTopicMessage = new RocketMqMessage<>("syncDelayLevelTopic", "延时消息，当前时间为:" + new Date());
        syncDelayLevelTopicMessage.setDelayMessageLevelEnum(DelayMessageLevelEnum.FIVE_SECONDS);
        MQSendResult<SendResult> sendResultMQSendResult = baseProducer.sendSyncDelayLevel(syncDelayLevelTopicMessage, String.class);
        return sendResultMQSendResult;
    }

    @Override
    public MQSendResult sendMessageInTransaction() {

        RocketMqMessage<String> messageInTransactionMessage = RocketMqMessage
                .buildJackMqMessage(String.class, "messageInTransactionTopic", "事务消息")
                .setHeaders("bit", "123")
                .setArg("123")
                .setTransactionId("ceshiyongshiwuID")
                .setHeaders(ProducerTxService.PRODUCER_TX_SERVICE_BEAN_NAME, SpringUtil.getDefaultBeanName(Default2ProducerTxServiceImpl.class));

        MQSendResult<TransactionSendResult> transactionSendResultMQSendResult = baseProducer.sendMessageInTransaction(messageInTransactionMessage, String.class);

        return transactionSendResultMQSendResult;
    }
}
